package com.hucais.sync.es2hive.service

import com.hucais.core.constant.Constants
import com.hucais.core.utils.DefaultPropertiesUtil
import com.hucais.etl.common.service.CommonQueryService
import org.apache.spark.SparkContext
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}
import org.elasticsearch.spark.sql.EsSparkSQL

object SnycOpenBooksService {
  def action(ssc: SparkContext, sparkSession: SparkSession): Unit = {
    val hdfsPath: String = "/hucaisdata/published/ods/ods_openbooks/"
    val dataDS = SnycOpenBooksService.getEsData(sparkSession)
    dataDS.cache()
    CommonQueryService.saveAsFileAbsPath(dataDS.coalesce(3).toDF(), hdfsPath, Constants.HIVE_SPLIT_STR, SaveMode.Overwrite)
  }

  private def getEsData(sparkSession: SparkSession): Dataset[EsData] = {
    import sparkSession.implicits._
    val esQuery =
      """
        |{
        |  "query": {
        |    "match_all": {}
        |  }
        |}
       """.stripMargin
    val tmpDS = EsSparkSQL.esDF(sparkSession, DefaultPropertiesUtil.get("ods.opendata"), esQuery).as[EsData]
    val colArray=Array("isbn", "month_sales")

    // 去重和解决ES获取的数据和Hive对应字段顺序不一致
    tmpDS.dropDuplicates(colArray).mapPartitions(partitions => {
      partitions.map(item => {
        EsData(
          item.isbn, item.book_name, item.selling_price, item.discount_rate, item.author,
          item.category, item.publishing_house, item.month_sales, item.year_sales, item.total_sales,
          item.book_list, item.channel_type, item.sale_type, item.sale_time, item.acquisition_time,
          item.acquisition_timestamp
        )
      })
    })
  }


  case class EsData(
                     isbn: String,
                     book_name: String,
                     selling_price: String,
                     discount_rate: String,
                     author: String,
                     category: String,
                     publishing_house: String,
                     month_sales: String,
                     year_sales: String,
                     total_sales: String,
                     book_list: String,
                     channel_type: String,
                     sale_type: String,
                     sale_time: String,
                     acquisition_time: String,
                     acquisition_timestamp: String
                   )
}
